package com.javafactroy.kunlong.bip.icp.netty.client.framework;


import com.javafactroy.kunlong.bip.icp.common.consts.IConfig;
import com.javafactroy.kunlong.bip.icp.message.data.netty.handler.client.NettyClientHandlerInitilizer;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.javafactroy.kunlong.bip.icp.common.config.BipConfig;
import com.javafactroy.kunlong.bip.icp.common.consts.EAppType;
import com.javafactroy.kunlong.bip.icp.context.AppContext;
import com.javafactroy.kunlong.bip.icp.message.base.iface.IMessage;

import java.util.Properties;
import java.util.concurrent.TimeUnit;


/**
 * @Description: Netty客户端
 * @Author:leijmdas
 * @Since:2019年9月26日下午8:54:59
 */
@Component
@Data
@Slf4j
public class NettyClientPromise {

    @Autowired
    BipConfig bipConfig;
    EAppType appType;

    public NettyClientPromise() {
    }

    private static volatile EventLoopGroup group;// = new NioEventLoopGroup();

    public void close() throws InterruptedException {
        getSocketChannel().close().sync();
        //getGroup().shutdownGracefully();
        //setGroup(null);
    }

    /**
     * @Fields socketChannel : 通道
     */
    private SocketChannel socketChannel;

    /**
     * @Fields clientHandlerInitilizer : 初始化
     */

    private NettyClientHandlerInitilizer initilizer;

    /**
     * @Description: 启动客户端
     * @Author: 雷建明
     * @Since: 2023:43:21
     */
    @SuppressWarnings("unchecked")
    public synchronized void start(EAppType appType) throws InterruptedException {
        this.appType = appType;
        log.info("****appType = {}", appType);

        initilizer = AppContext.getBean(NettyClientHandlerInitilizer.class);
        if (group == null) {
            synchronized (NettyClientPromise.class) {
                if (group == null) {
                    group = new NioEventLoopGroup();
                }
            }
        }

        String hostIp = bipConfig.getServerIp();
        int port = bipConfig.getServerPort();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                // 指定Channel
                .channel(NioSocketChannel.class)
                // 服务端地址
                .remoteAddress(hostIp, port)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(initilizer);

        // 连接
        ChannelFuture channelFuture = bootstrap.connect().sync();
        socketChannel = (SocketChannel) channelFuture.channel();
        //客户端断线重连逻辑
        channelFuture.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("连接Netty服务端成功...");
                } else {

                    log.info("连接Netty服务端失败，进行断线重连...");
                    final EventLoop loop = future.channel().eventLoop();
                    loop.schedule(new Runnable() {
                        @SneakyThrows
                        @Override
                        public void run() {
                            log.info("连接正在重试...");
                            start(appType);
                        }
                    }, 30, TimeUnit.SECONDS);
                }
            }
        });


    }


    /**
     * @param message
     * @Description: 消息发送
     * @Author: 雷建明
     * @Since: 2019年9月12日下午5:08:47
     */


    /**
     * @param message
     * @Description: 发送同步消息
     * @Author: 雷建明
     * @Since: 2019年9月12日下午5:08:47
     */
    public String sendSyncMsg(String message, DxpMsgSyncFuture<String> syncFuture) {

        String result = "";
        ByteBuf byteBuf = Unpooled.copiedBuffer("", CharsetUtil.UTF_8);

        try {

            ChannelFuture future = socketChannel.writeAndFlush(byteBuf);
            future.addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                    if (future.isSuccess()) {
                        System.out.println("===========发送成功");
                    } else {
                        System.out.println("------------------发送失败");
                    }
                }
            });

            // 等待 8 秒
            result = syncFuture.get(30, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return result;
    }

    public IMessage sendPromise(IMessage message) {
        initilizer.setPromise(socketChannel.eventLoop());

        log.info("sendPromise ={}", message.toString());
        try {
            ChannelFuture future = socketChannel.writeAndFlush(message);
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {

                    if (future.isSuccess()) {
                        log.info("===========发送成功！");
                    } else {
                        log.error("-----------发送失败！！！");
                    }
                }
            });

            initilizer.getPromise().sync();
            return initilizer.getPromise().getNow();//(10, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getCause());
        }

    }

    //DefaultPromise<IMessage> promise
    public IMessage sendSyncMsg(IMessage message, DxpMsgSyncFuture<IMessage> syncFuture) {

        log.info("sendSyncMsg ={}", message.toString());
        try {

            ChannelFuture future = socketChannel.writeAndFlush(message);
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                    if (future.isSuccess()) {
                        log.info("===========发送成功！");
                    } else {
                        log.error("-----------发送失败！！！");

                        DxpMsgSyncFutureMngr.ackSyncMsgErr(message.getMessageId());
                        start(appType);
                    }
                }
            });
            return syncFuture.get(10, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getCause());
        }


    }

    public void setIpPort(String ip, int port) {
        if (bipConfig == null) {
            bipConfig = new BipConfig();
            bipConfig.setProps(new Properties());
        }
        if (bipConfig.getProps() == null) {
            bipConfig.setProps(new Properties());
        }
        bipConfig.getProps().setProperty("server.ip", ip);
        bipConfig.getProps().setProperty("server.port", String.valueOf(port));
    }

    public IMessage send(IMessage message) {
        return sendSyncMsg(message);
    }

    public IMessage sendSyncMsg(IMessage message) {

        DxpMsgSyncFuture<IMessage> syncFuture = new DxpMsgSyncFuture<IMessage>();
        // 放入缓存中
        DxpMsgSyncFutureMngr.put(message.getMessageId() + "", syncFuture);
        // 发送同步消息
        IMessage iMessage = sendSyncMsg(message, syncFuture);
        if (iMessage.getMsgState() != null && iMessage.getMsgState() == IConfig.INettyConst.MSG_State.MSG_SEND_ERR) {
            iMessage = sendSyncMsg(message, syncFuture);
        }
        return iMessage;

    }


}
